package com.example.messagequeue.mqserver.datacenter;

import com.example.messagequeue.commom.MqException;
import com.example.messagequeue.mqserver.core.Binding;
import com.example.messagequeue.mqserver.core.Exchange;
import com.example.messagequeue.mqserver.core.MSGQueue;
import com.example.messagequeue.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/*
 * 使用这个类来统一管理内存中的所有数据
 * 注意线程安全
 */
public class MemoryDataCenter {
    //key: exchangeName value: Exchange 对象
    private ConcurrentHashMap<String, Exchange> exchangesMap = new ConcurrentHashMap<>();
    //key: queueName value: MSGQueue 对象
    private ConcurrentHashMap<String, MSGQueue> queuesMap = new ConcurrentHashMap<>();
    //key: messageId value: Message 对象
    private ConcurrentHashMap<String, Message> messagesMap = new ConcurrentHashMap<>();
    //第一个key：exchangName， 第二个key：queueName
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    //key: queueName value:message 链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessagesMap = new ConcurrentHashMap<>();
    //第一个key：queueName， 第二个key：messageId
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessagesAckMap = new ConcurrentHashMap<>();

    public void insertExchang(Exchange exchange){
        exchangesMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功 exchangeName = " + exchange.getName());
    }

    public Exchange getExchange(String exchangeName){
        return exchangesMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName){
        exchangesMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功 exchangeName = " + exchangeName);
    }

    public void insertQueue(MSGQueue queue){
        queuesMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 新队列添加成功 queueName = " + queue.getName());
    }

    public MSGQueue getQueue(String queueName){
        return queuesMap.get(queueName);
    }

    public void deleteQueue(String queueName){
        queuesMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功 queueName = " + queueName);
    }

    public void insertBinding(Binding binding) throws MqException {
//        ConcurrentHashMap<String, Binding> bindingMap =  bindingsMap.get(binding);
//        if (bindingMap == null){
//            bindingMap = new ConcurrentHashMap<>();
//            bindingsMap.put(binding.getExchangeName(), bindingMap);
//        }
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k ->new ConcurrentHashMap<>());

        synchronized (bindingMap){
            if(bindingMap.get(binding.getQueueName()) != null){
                throw new MqException("[MemoryDataCenter] 绑定已经存在 exchangeName = " + binding.getExchangeName() +
                        " queueName" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter] 新绑定添加成功 exchangeName = " + binding.getExchangeName() +
                " queueName" + binding.getQueueName());
    }

    //1.根据exchangName和queueName获取确定的binding
    //2.根据exchangName获取所有的binding
    public Binding getBinding(String exchangeName, String queueName){
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null){
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBinding(String exchangeName){
        return bindingsMap.get(exchangeName);
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if(bindingMap == null){
            throw new MqException("[MemoryDataCenter] 绑定不存在 exchangeName = " + binding.getExchangeName() +
                    " queueName" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功 exchangeName = " + binding.getExchangeName() +
                " queueName" + binding.getQueueName());
    }

    public void addMessage(Message message){
        messagesMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 新消息添加成功 messageId = " + message.getMessageId());
    }

    public void removeMessage(String messageId){
        messagesMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息删除成功 messageId = " + messageId);
    }

    public Message getMessage(String messageId){
        return messagesMap.get(messageId);
    }

    public void sendMessage(MSGQueue queue, Message message){
        //把消息放到对应的队列数据结构中
        //先根据队列的名字，找到该队列对应的消息链表
        LinkedList<Message> messages = queueMessagesMap.computeIfAbsent(queue.getName(),
                k -> new LinkedList<>());
        synchronized (messages){
            //再把数据加到messages中
            messages.add(message);
        }
        //再将message往消息中心加入，即使message已经在消息中心存在，重复插入也不会造成改变
        addMessage(message);
        System.out.println("[MemoryDataCenter] 消息插入到队列中 messageId = " + message.getMessageId());
    }

    public Message pollMessage(String queueName){
        LinkedList<Message> messages = queueMessagesMap.get(queueName);
        if (messages == null){
            return null;
        }
        synchronized (messages){
            if (messages.size() == 0){
                return null;
            }
            Message message = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出 messageId = " + message.getMessageId());
            return message;
        }
    }

    public int getMessageCount(String queueName){
        LinkedList<Message> messages = queueMessagesMap.get(queueName);
        if (messages == null){
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }

    //添加未确认消息
    public void addMessageWaitAck(String queueName, Message message){
        ConcurrentHashMap<String, Message> messageHashMap = queueMessagesAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 消息进入待确认对列 messageId = " + message.getMessageId());
    }

    //删除未确认消息
    public void removeMessageWaitAck(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messageHashMap = queueMessagesAckMap.get(queueName);
        if (messageHashMap == null){
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认对列取出 messageId = " + messageId);
    }

    //获取指定的未确认消息
    public Message getMessageWaitAck(String queueName, String messageId){
        ConcurrentHashMap<String, Message> messageHashMap = queueMessagesAckMap.get(queueName);
        if (queueName == null){
            return null;
        }
        return messageHashMap.get(messageId);
    }

    //这个方法就是从硬盘上读取数据，把硬盘中之前持久化储存的各个维度的数据恢复到内存中
    public void recovery(DisDataCenter disDataCenter) throws IOException, MqException, ClassNotFoundException {
        //清空之前的数据
        exchangesMap.clear();
        queuesMap.clear();
        bindingsMap.clear();
        queueMessagesMap.clear();
        messagesMap.clear();
        //1恢复交换机数据
        List<Exchange> exchanges = disDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges){
            exchangesMap.put(exchange.getName(), exchange);
        }
        //2恢复队列数据
        List<MSGQueue> queues = disDataCenter.selectAllQueues();
        for (MSGQueue queue : queues){
            queuesMap.put(queue.getName(),queue);
        }
        //3恢复绑定数据
        List<Binding> bindings = disDataCenter.selectAllBindings();
        for (Binding binding : bindings){
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }
        //4恢复消息数据
        for (MSGQueue queue : queues){
            LinkedList<Message> messages = disDataCenter.loadAllMessage(queue.getName());
            queueMessagesMap.put(queue.getName(), messages);
            for (Message message : messages){
                messagesMap.put(message.getMessageId(), message);
            }
        }
        //注意，针对未确认的消息这部分内存中的数据，不需要从硬盘恢复，之前考虑硬盘存储的时候，也没设定这一块
        //一旦在等待ack的过程中，服务器重启了，此时这些违背确认的消息就恢复成未被取走的消息
        //这个消息在硬盘上存储时，就当做未被取走

    }

}
